package com.gitee.kamismile.stone.rpc.api.invocation.support;

import com.gitee.kamismile.stone.rpc.common.dto.StoneAppResponse;
import com.gitee.kamismile.stone.rpc.common.engine.StoneRpcContext;
import com.gitee.kamismile.stone.rpc.common.exception.RPCException;
import com.gitee.kamismile.stone.rpc.common.invocation.IStoneAsyncResponse;
import com.gitee.kamismile.stone.rpc.common.support.StoneURL;
import org.apache.commons.lang.exception.ExceptionUtils;
import  org.slf4j.LoggerFactory;
import org.slf4j.Logger;

import java.util.Objects;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

public class StoneInvokerUtils {
    protected static final Logger logger = LoggerFactory.getLogger(StoneInvokerUtils.class);

    /**
     * it's a internal API, caller make sure already invoked SCBEngine.ensureStatusUp
     *
     * @param stoneInvocation
     * @param asyncResp
     */
    public static void reactiveInvoke(StoneInvocation stoneInvocation, IStoneAsyncResponse asyncResp) {
        StoneAppResponse response = new StoneAppResponse();

        try {
            stoneInvocation.setSync(false);

            stoneInvocation.next(ar -> {
                response.setData(ar.getData());
                response.setAttachments(ar.getAttachments());
                response.setException(ar.getException());
                asyncResp.handle(response);
            });
        } catch (Throwable e) {
            if(e instanceof RPCException){
                decrementWeight(-80);
            }
            //if throw exception,we can use 500 for status code ?
            logger.error("invoke failed, {}", ExceptionUtils.getFullStackTrace(e));
            response.setException(e);
            asyncResp.handle(response);
        } finally {
            StoneRpcContext.removeContext();
        }
    }

    public static StoneAppResponse innerSyncInvoke(StoneInvocation invocation) {
        StoneAppResponse response = new StoneAppResponse();
        try {
            CountDownLatch countDownLatch = new CountDownLatch(1);
            invocation.next(ar -> {
                response.setData(ar.getData());
                response.setAttachments(ar.getAttachments());
                response.setException(ar.getException());
                countDownLatch.countDown();
            });
            if (invocation.getTimeout() > 0) {
                countDownLatch.await(invocation.getTimeout(), TimeUnit.MILLISECONDS);
            } else {
                countDownLatch.await();
            }
        } catch (Throwable e) {
            if(e instanceof RPCException){
                decrementWeight(-80);
            }
            response.setSuccess(false);
            response.setException(e);
            String msg = String.format("invoke failed, %s", ExceptionUtils.getFullStackTrace(e));
            logger.error(msg);
        } finally {
            StoneRpcContext.removeContext();
        }

        return response;
    }


    private static void decrementWeight(int delta) {
        StoneURL chooseStoneURL = StoneRpcContext.getLocalContext().getChooseStoneURL();
        if (!Objects.isNull(chooseStoneURL)) {
            if (chooseStoneURL.getWeight() > 50 || delta < -10) {
                chooseStoneURL.getWeightAtomic().addAndGet(delta);
            } else {
                chooseStoneURL.getAndDecrement();
            }
        }
    }

}
